#include "redis_server.h"

static bool IsHexDigit(char ch) {
  return (ch >= '0' && ch <= '9') || (ch >= 'a' && ch <= 'f') ||
         (ch >= 'A' && ch < 'F');
}

static int HexDigitToInt32(char ch) {
  if (ch <= '9' && ch >= '0') {
    return ch - '0';
  } else if (ch <= 'F' && ch >= 'A') {
    return ch - 'A';
  } else if (ch <= 'f' && ch >= 'a') {
    return ch - 'a';
  } else {
    return 0;
  }
}

static int Split2args(const std::string& buf, RedisCmdArgs& argv) {
  const char* p = buf.data();
  std::string arg;

  while (1) {
    // skip blanks
    while (*p && isspace(*p)) p++;
    if (*p) {
      // get a token
      int inq = 0;   // set to 1 if we are in "quotes"
      int insq = 0;  // set to 1 if we are in 'single quotes'
      int done = 0;

      arg.clear();
      while (!done) {
        if (inq) {
          if (*p == '\\' && *(p + 1) == 'x' && IsHexDigit(*(p + 2)) &&
              IsHexDigit(*(p + 3))) {
            unsigned char byte =
                HexDigitToInt32(*(p + 2)) * 16 + HexDigitToInt32(*(p + 3));
            arg.append(1, byte);
            p += 3;
          } else if (*p == '\\' && *(p + 1)) {
            char c;

            p++;
            switch (*p) {
              case 'n':
                c = '\n';
                break;
              case 'r':
                c = '\r';
                break;
              case 't':
                c = '\t';
                break;
              case 'b':
                c = '\b';
                break;
              case 'a':
                c = '\a';
                break;
              default:
                c = *p;
                break;
            }
            arg.append(1, c);
          } else if (*p == '"') {
            /* closing quote must be followed by a space or
             * nothing at all. */
            if (*(p + 1) && !isspace(*(p + 1))) {
              argv.clear();
              return -1;
            }
            done = 1;
          } else if (!*p) {
            // unterminated quotes
            argv.clear();
            return -1;
          } else {
            arg.append(1, *p);
          }
        } else if (insq) {
          if (*p == '\\' && *(p + 1) == '\'') {
            p++;
            arg.append(1, '\'');
          } else if (*p == '\'') {
            /* closing quote must be followed by a space or
             * nothing at all. */
            if (*(p + 1) && !isspace(*(p + 1))) {
              argv.clear();
              return -1;
            }
            done = 1;
          } else if (!*p) {
            // unterminated quotes
            argv.clear();
            return -1;
          } else {
            arg.append(1, *p);
          }
        } else {
          switch (*p) {
            case ' ':
            case '\n':
            case '\r':
            case '\t':
            case '\0':
              done = 1;
              break;
            case '"':
              inq = 1;
              break;
            case '\'':
              insq = 1;
              break;
            default:
              arg.append(1, *p);
              break;
          }
        }
        if (*p) p++;
      }
      argv.push_back(arg);
    } else {
      return 0;
    }
  }
}

RedisConn::RedisConn()
    : req_type_(0),
      multibulk_len_(0),
      bulk_len_(-1),
      last_read_pos_(0),
      next_parse_pos_(0) {}

RedisConn::~RedisConn() {}

void RedisConn::OnConnectionCallback(
    const std::shared_ptr<zrpc::TcpConnection>& conn) {
  if (conn->Connected()) {
    rbuf_ = conn->IntputBuffer();
    wbuf_ = conn->OutputBuffer();
    conn_ = conn;
    conn->SetMessageCallback(std::bind(&RedisConn::OnMessageCallback, this,
                                       std::placeholders::_1,
                                       std::placeholders::_2));
  }
}

void RedisConn::OnMessageCallback(
    const std::shared_ptr<zrpc::TcpConnection>& conn, zrpc::Buffer* buf) {
  last_read_pos_ += buf->ReadableBytes();

  ReadStatus ret = ProcessInputBuffer();
  if (ret == kReadAll) {
    if (wbuf_->ReadableBytes() > 0) {
      conn_->SendPipe();
    }

    rbuf_->Retrieve(next_parse_pos_);
    last_read_pos_ -= next_parse_pos_;
    assert(buf->ReadableBytes() == last_read_pos_);
    next_parse_pos_ = 0;
  }
}

ReadStatus RedisConn::ProcessInputBuffer() {
  ReadStatus ret;
  while (next_parse_pos_ < last_read_pos_) {
    if (!req_type_) {
      if (rbuf_->Peek()[next_parse_pos_] == '*') {
        req_type_ = REDIS_REQ_MULTIBULK;
      } else {
        req_type_ = REDIS_REQ_INLINE;
      }
    }

    if (req_type_ == REDIS_REQ_INLINE) {
      ret = ProcessInlineBuffer();
      if (ret != kReadAll) {
        return ret;
      }
    } else if (req_type_ == REDIS_REQ_MULTIBULK) {
      ret = ProcessMultibulkBuffer();
    } else {
      return kParseError;
    }

    if (ret != kReadAll) {
      return ret;
    }

    if (!argv_.empty()) {
      if (redis_cmd_args_callback_) {
        redis_cmd_args_callback_(argv_, conn_, wbuf_);
      }

      OnCommandCallback();
    }

    argv_.clear();
    req_type_ = 0;
    multibulk_len_ = 0;
    bulk_len_ = -1;
  }

  return kReadAll;
}

ReadStatus RedisConn::ProcessMultibulkBuffer() {
  int pos = 0;
  if (multibulk_len_ == 0) {
    pos = FindNextSeparators();
    if (pos != -1) {
      if (Strtoll(pos, &multibulk_len_)) {
        return kParseError;
      }

      next_parse_pos_ = pos + 1;
      argv_.clear();

      if (next_parse_pos_ > last_read_pos_) {
        return kReadHalf;
      }
    } else {
      return kReadHalf;
    }
  }

  while (multibulk_len_) {
    if (bulk_len_ == -1) {
      pos = FindNextSeparators();
      if (pos != -1) {
        if (rbuf_->Data()[next_parse_pos_] != '$') {
          return kParseError;
        }

        if (Strtoll(pos, &bulk_len_)) {
          return kParseError;
        }
        next_parse_pos_ = pos + 1;
      }

      if (pos == -1 || next_parse_pos_ > last_read_pos_) {
        return kReadHalf;
      }
    }

    if (last_read_pos_ - next_parse_pos_ + 1 < bulk_len_ + 2) {
      break;
    } else {
      argv_.emplace_back(rbuf_->Data() + next_parse_pos_, bulk_len_);
      next_parse_pos_ = next_parse_pos_ + bulk_len_ + 2;
      bulk_len_ = -1;
      multibulk_len_--;
    }
  }

  if (multibulk_len_ == 0) {
    return kReadAll;
  } else {
    return kReadHalf;
  }
}

ReadStatus RedisConn::ProcessInlineBuffer() {
  int pos, ret;
  pos = FindNextSeparators();
  if (pos == -1) {
    return kReadHalf;
  }

  std::string read_buf(rbuf_->Data() + next_parse_pos_,
                       pos + 1 - next_parse_pos_);
  argv_.clear();
  ret = Split2args(read_buf, argv_);
  next_parse_pos_ = pos + 1;
  return ret == -1 ? kParseError : kReadAll;
}

int RedisConn::FindNextSeparators() {
  if (next_parse_pos_ > last_read_pos_) {
    return -1;
  }

  int pos = next_parse_pos_;
  while (pos <= last_read_pos_) {
    if (rbuf_->Peek()[pos] == '\n') {
      return pos;
    }
    pos++;
  }
  return -1;
}

int RedisConn::Strtoll(int pos, int32_t* value) {
  assert(pos > next_parse_pos_);

  char* begin = rbuf_->Data() + next_parse_pos_ + 1;
  char* end = rbuf_->Data() + pos - next_parse_pos_ - 2;
  *value = std::strtoll(begin, &end, 10);
  if ((errno == ERANGE && (*value == ULLONG_MAX)) ||
      (errno != 0 && *value == 0)) {
    return 1;
  }
  return 0;
}

void RedisConn::OnCommandCallback() {
  //printf("get redis message ");
  for (int i = 0; i < argv_.size(); i++) {
    //printf("%s ", argv_[i].c_str());
  }
  //printf("\n");

  if ((argv_[0] == "set" || argv_[0] == "SET") && (argv_.size() == 3)) {
    wbuf_->Append("+OK\r\n", 5);
    db_set_commands_[argv_[1]] = argv_[2];
  } else if ((argv_[0] == "get" || argv_[0] == "GET") && (argv_.size() == 2)) {
    auto iter = db_set_commands_.find(argv_[1]);
    if (iter != db_set_commands_.end()) {
      const std::string& val = iter->second;
      wbuf_->Append("*1\r\n$", 5);
      wbuf_->Append(std::to_string(val.size()));
      wbuf_->Append("\r\n", 2);
      wbuf_->Append(val.c_str(), val.size());
      wbuf_->Append("\r\n", 2);
    } else {
      wbuf_->Append("$-1\r\n", 4);
    }
  } else {
    wbuf_->Append("+OK\r\n", 5);
    //std::string reply_str_err = "-ERR unknown command `" + argv_[0] + "`" +
    //                            ", with args beginning with:";
    //for (int i = 1; i < argv_.size(); i++) {
    //  reply_str_err += " `";
    //  reply_str_err += argv_[i];
    //  reply_str_err += "`";
    //  reply_str_err += ",";
    //}

    //reply_str_err += "\r\n";
    //wbuf_->Append(reply_str_err.c_str(), reply_str_err.size());
  }
}

RedisServer::RedisServer(const std::string& ip, int16_t port)
    : tcp_server_(&event_loop_, ip, port, nullptr) {
  tcp_server_.SetAcceptCallback(
      std::bind(&RedisServer::OnAcceptCallback, this, std::placeholders::_1));
  tcp_server_.SetConnectionCallback(std::bind(
      &RedisServer::OnConnectionCallback, this, std::placeholders::_1));
  tcp_server_.Start();
}

RedisServer ::~RedisServer() {}

void RedisServer::Run() { event_loop_.Run(); }

void RedisServer::OnConnectionCallback(
    const std::shared_ptr<zrpc::TcpConnection>& conn) {
  if (conn->Connected()) {
    //printf("from client connect :%d\n", conn->GetSockfd());
    std::shared_ptr<RedisConn> redis_conn(new RedisConn());
    redis_conn->OnConnectionCallback(conn);
    redis_conns_[conn->GetSockfd()] = redis_conn;
  } else {
    //printf("from client disconnect :%d\n", conn->GetSockfd());
    size_t n = redis_conns_.erase(conn->GetSockfd());
    assert(n == 1);
  }
}

void RedisServer::OnAcceptCallback(const int32_t sockfd) {}
int main() {
  RedisServer redis_server("127.0.0.1", 6379);
  redis_server.Run();
}
